use std::{collections::VecDeque, future::poll_fn, sync::Arc, time::Duration};

use crate::{
    config::http::{
        core_conf::HttpCoreRuntimeConf, main::get_http_main_runtime_conf,
        site_mgr::HttpSiteRuntimeConfManager,
    },
    error::*,
    http::base::{
        conn::{HttpAddressSiteRuntimeConfInfo, HttpStream},
        header_name::HeaderName,
        headers::HttpHeaders,
        parse::HttpResult,
        status_code::{http_body_allow_status, StatusCode},
        Request, Response,
    },
};
use bytes::Bytes;
use h2::{
    server::{Builder as H2Builder, SendResponse},
    Reason, RecvStream, SendStream,
};
use http::{Method, Version};
use log::debug;
use tokio::time::timeout;

use crate::http::base::conn::BoxHttpConn;

type H2Connection = h2::server::Connection<BoxHttpConn, Bytes>;

#[allow(dead_code)]
#[inline]
pub async fn handshake(
    hc: BoxHttpConn,
    h2_builder: Option<&H2Builder>,
) -> Result<H2Connection> {
    let h2_builder = match h2_builder {
        Some(h2_builder) => h2_builder,
        None => &H2Builder::default(),
    };

    h2_builder.handshake(hc).await.if_err(
        ErrorType::HandshakeFailed,
        "while h2 handshakeing wtih client",
    )
}

#[allow(dead_code)]
pub struct ServerSession {
    hc: BoxHttpConn,
    req_body_reader: RecvStream,
    send_resp: SendResponse<Bytes>,

    // config
    body_read_timeout_ms: usize,
    body_write_timedout_ms: usize,

    // request
    req: Request,
    expect_trailers: usize,
    trailers: Option<HttpHeaders>,
    read_done: bool,

    // response
    pub header_send: bool,
    resp_no_body: bool,
    expect_write_trailers: usize,
    body_writer: Option<SendStream<Bytes>>,
    write_body_done: bool,
    write_trailers_done: bool,

    // response
    stream_ended: bool,
}

#[allow(dead_code)]
impl ServerSession {
    pub fn new(mut hc: BoxHttpConn) -> Self {
        // 将recv_req和send_resp的所有权转移至server session.
        let (recv_req, send_resp) = match &mut hc.raw_conn {
            HttpStream::H2Stream(h2_stream) => (
                h2_stream.recv_req.take().unwrap(),
                h2_stream.send_resp.take().unwrap(),
            ),
            _ => unreachable!("expect h2 stream, but got other"),
        };

        let (req_parts, req_body_reader) = recv_req.into_parts();
        let mut req = Request::new();
        req.method = req_parts.method;
        req.uri = req_parts.uri.clone();
        req.version = Version::HTTP_2;

        for (name, value) in req_parts.headers.iter() {
            req.headers.append_header_with_kv(
                Bytes::copy_from_slice(name.as_str().as_bytes()),
                Bytes::copy_from_slice(value.as_bytes()),
            );
        }

        ServerSession {
            hc,
            req_body_reader,
            send_resp,

            body_read_timeout_ms: 0,
            body_write_timedout_ms: 0,

            req,
            expect_trailers: 0,
            trailers: None,
            read_done: false,

            header_send: false,
            resp_no_body: false,
            expect_write_trailers: 0,
            body_writer: None,
            write_body_done: false,
            write_trailers_done: false,

            stream_ended: false,
        }
    }

    pub fn get_hc(&mut self) -> &mut BoxHttpConn {
        &mut self.hc
    }

    // 该方法有两种返回值:
    // - Ok(Some(req)): process request success.
    // - Err(err): process request error.
    pub fn read_request(&mut self) -> Result<Option<&mut Request>> {
        // 无需这样做, h2的每个请求都是一个新的hc.
        // if let Some(info) = &mut self.hc.site_conf_info {
        //     info.valid = false;
        // }
        self.set_session_conf()?;

        self.process_request()?;

        Ok(Some(&mut self.req))
    }

    pub fn request(&mut self) -> &mut Request {
        &mut self.req
    }

    pub async fn read_request_body_helper(
        &mut self,
        bufs: &mut VecDeque<Bytes>,
    ) -> Result<HttpResult> {
        let read_event = self.req_body_reader.data();
        let read_event = timeout(
            Duration::from_millis(self.body_read_timeout_ms as u64),
            read_event,
        );

        let data = read_event
            .await
            .if_err(
                ErrorType::ReadTimedout,
                "while read request body in h2 stream",
            )?
            .transpose()
            .if_err(
                ErrorType::ReadError,
                "while reading downstream request body",
            )?;

        if let Some(data) = data {
            let data_len = data.len();
            bufs.push_back(data);
            // 必须释放flow_control, h2c才可以继续读取更多的数据.
            self.req_body_reader
                .flow_control()
                .release_capacity(data_len)
                .if_err(ErrorType::H2Error, "h2 flow control release failed")?;
            return Ok(HttpResult::Partial);
        }

        // data is None, read request body done.
        let trailers = match self.req_body_reader.trailers().await.if_err(
            ErrorType::H2Error,
            "while reading downstream request trailers",
        )? {
            Some(t) => t,
            None => {
                self.read_done = true;
                return Ok(HttpResult::Complete);
            }
        };

        let mut headers = HttpHeaders::new();
        for (name, value) in trailers.iter() {
            headers.append_header_with_kv(
                Bytes::copy_from_slice(name.as_str().as_bytes()),
                Bytes::copy_from_slice(value.as_bytes()),
            );
        }

        if headers.len() != self.expect_trailers {
            return Error::e_explain(
                ErrorType::H2Error,
                format!(
                    "expect {} trailer headers, but got {}",
                    self.expect_trailers,
                    headers.len()
                ),
            );
        }

        self.trailers = Some(headers);
        self.read_done = true;

        Ok(HttpResult::Complete)
    }

    pub async fn read_request_body(
        &mut self,
        bufs: &mut VecDeque<Bytes>,
    ) -> Result<HttpResult> {
        for _ in 0..2 {
            let res = self.read_request_body_helper(bufs).await?;
            if res == HttpResult::Complete {
                return Ok(HttpResult::Complete);
            }
        }

        Ok(HttpResult::Partial)
    }

    pub fn expect_trailers_num(&self) -> usize {
        self.expect_trailers
    }

    pub fn get_req_trailer_headers(&self) -> Result<Option<&HttpHeaders>> {
        Ok(self.trailers.as_ref())
    }

    pub async fn send_resp_header(
        &mut self,
        resp: &mut Response,
    ) -> Result<()> {
        // 我们不响应非101的其他1XX状态码.
        if resp.status_code >= 100
            && resp.status_code <= 199
            && resp.status_code != StatusCode::HTTP_SWITCHING_PROTOCOLS
        {
            debug!("ignoring informational header");
            return Ok(());
        }

        // header already send.
        if self.header_send {
            return Error::e_explain(
                ErrorType::InternalError,
                "response header already send done",
            );
        }
        self.header_send = true;

        if self.req.method == Method::HEAD {
            self.resp_no_body = true;
        }

        // fix send response header
        if resp.status_code < StatusCode::HTTP_STATUS_START
            || resp.status_code > StatusCode::HTTP_STATUS_END
        {
            resp.status_code = StatusCode::HTTP_INTERNAL_SERVER_ERROR;
        }

        if !http_body_allow_status(resp.status_code) {
            self.resp_no_body = true;
        }

        // FIXME: H2的库不支持响应自定义的status_reason.
        // if resp.status_reason.is_empty() {
        //     resp.status_reason =
        //         get_status_reason(resp.status_code).to_string();
        // }

        // remove other h1 hop headers that cannot be present in H2
        // https://httpwg.org/specs/rfc7540.html#n-connection-specific-header-fields
        resp.headers.del_header_all(HeaderName::TRANSFER_ENCODING);
        resp.headers.del_header_all(HeaderName::CONNECTION);
        resp.headers.del_header_all(HeaderName::UPGRADE);
        resp.headers.del_header_all(b"keep-alive");
        resp.headers.del_header_all(b"proxy-connection");

        let mut parts = http::response::Builder::new()
            .body(())
            .unwrap()
            .into_parts()
            .0;

        parts.status = http::StatusCode::from_u16(resp.status_code)
            .if_err(ErrorType::InternalError, "invalid response status")?;

        for h in resp.headers.iter() {
            let name = h.get_name();
            let value = h.get_value();

            match name.len() {
                7 => {
                    if name.eq_ignore_ascii_case(HeaderName::TRAILER) {
                        let iter = value.split(|c| *c == b',');
                        for mut v in iter {
                            v = v.trim_ascii();
                            if !v.is_empty() {
                                self.expect_write_trailers += 1;
                            }
                        }
                    }
                }

                _ => {
                    // Nothing.
                }
            }

            let header_name = http::HeaderName::from_bytes(name).if_err(
                ErrorType::H2Error,
                format!(
                    "Failed to convert {} to header name",
                    String::from_utf8_lossy(h.get_name())
                ),
            )?;

            let header_value = http::HeaderValue::from_bytes(value).if_err(
                ErrorType::H2Error,
                format!(
                    "Failed to convert {} to header value",
                    String::from_utf8_lossy(h.get_value())
                ),
            )?;

            parts.headers.insert(header_name, header_value);
        }

        let resp = http::Response::from_parts(parts, ());
        let body_writer = self
            .send_resp
            .send_response(resp, self.resp_no_body)
            .if_err(
                ErrorType::WriteError,
                "while writing h2 response to downstream",
            )?;

        self.body_writer = Some(body_writer);

        if self.resp_no_body {
            self.write_body_done = true;
            self.expect_write_trailers = 0;
            self.stream_ended = true;
        }

        Ok(())
    }

    pub async fn write_resp_body(
        &mut self,
        bufs: &mut VecDeque<Bytes>,
        done: bool,
    ) -> Result<HttpResult> {
        if self.resp_no_body {
            return Ok(HttpResult::Complete);
        }

        if self.body_writer.is_none() {
            return Error::e_explain(
                ErrorType::InternalError,
                "body writer no inited",
            );
        }

        let end_stream = done && self.expect_write_trailers == 0;

        while let Some(data) = bufs.pop_front() {
            let end = end_stream && bufs.is_empty();
            self.write_resp_body_helper(data, end).await?;
        }

        if end_stream {
            self.stream_ended = true;
        }

        if done {
            self.write_body_done = true;
            return Ok(HttpResult::Complete);
        }

        Ok(HttpResult::Partial)
    }

    pub fn expect_write_trailers_num(&mut self) -> usize {
        self.expect_write_trailers
    }

    pub async fn write_resp_trailers(
        &mut self,
        headers: &HttpHeaders,
    ) -> Result<()> {
        if self.resp_no_body {
            return Ok(());
        }

        if !self.write_body_done {
            return Error::e_explain(
                ErrorType::InternalError,
                "response body has not been sent completely",
            );
        }

        if self.write_trailers_done {
            return Error::e_explain(
                ErrorType::InternalError,
                "response trailers already write",
            );
        }

        if self.expect_write_trailers == 0 {
            self.write_trailers_done = true;
            return Ok(());
        }

        if headers.len() != self.expect_trailers {
            return Error::e_explain(
                ErrorType::InvalidTrailers,
                format!(
                    "expect resp trailers header num: {}, but got none",
                    self.expect_trailers
                ),
            );
        }

        let body_writer = self.body_writer.as_mut().unwrap();
        let mut header_map = http::HeaderMap::new();
        for h in headers.iter() {
            let name = h.get_name();
            let value = h.get_value();

            let header_name = http::HeaderName::from_bytes(name).if_err(
                ErrorType::H2Error,
                format!(
                    "Failed to convert {} to header name",
                    String::from_utf8_lossy(h.get_name())
                ),
            )?;
            let header_value = http::HeaderValue::from_bytes(value).if_err(
                ErrorType::H2Error,
                format!(
                    "Failed to convert {} to header value",
                    String::from_utf8_lossy(h.get_value())
                ),
            )?;

            header_map.insert(header_name, header_value);
        }

        body_writer
            .send_trailers(header_map)
            .if_err(ErrorType::H2StreamError, "while h2 send resp trailers")?;
        // send_trailers之后, stream被隐式关闭.
        self.stream_ended = true;
        self.write_trailers_done = true;
        Ok(())
    }

    pub fn write_resp_done(&self) -> bool {
        if !self.write_body_done {
            return false;
        }

        if self.expect_write_trailers > 0 && !self.write_trailers_done {
            return false;
        }

        true
    }

    pub async fn finalize(&mut self, _abort: bool) {
        // FIXME: request_body字节数统计.
        // 客户端发送了大量的request body, 我们并没有保证一直读取request body直到完毕.
        // 在某些异常场景中, 这会导致request_body_bytes比实际的要小.
        // 貌似没有更好的处理办法?
        // 在Nginx中对于无效的frame data, 也是单纯的丢弃, 无法正确统计到对应的请求级别?

        if self.stream_ended {
            return;
        }

        if let Some(body_writer) = self.body_writer.as_mut() {
            body_writer.send_reset(Reason::REFUSED_STREAM);
        } else {
            // 在发送response header或之前发生错误.
            self.send_resp.send_reset(Reason::REFUSED_STREAM);
        }

        self.stream_ended = true;
    }
}

impl ServerSession {
    #[inline]
    fn set_session_conf(&mut self) -> Result<()> {
        if let Some(host_header) = self.req.headers.get_header(HeaderName::HOST)
        {
            let listener_address =
                self.hc.listener.clone().unwrap().get_conf().address.clone();

            let host_str =
                String::from_utf8_lossy(host_header.get_value()).to_string();

            let address_site_conf =
                HttpSiteRuntimeConfManager::lookup_address_site_conf(
                    &host_str,
                    listener_address.as_bytes(),
                )?;

            match address_site_conf {
                Some(conf) => {
                    self.hc.address_site_conf_info =
                        Some(HttpAddressSiteRuntimeConfInfo {
                            conf: conf.clone(),
                            valid: true,
                        });
                    self.update_session_conf(&conf.core_runtime_conf);
                }

                None => {
                    let main_conf = get_http_main_runtime_conf();
                    self.update_session_conf(&main_conf.core_runtime_conf);
                }
            }
        }

        Ok(())
    }

    fn update_session_conf(
        &mut self,
        core_runtime_conf: &Arc<HttpCoreRuntimeConf>,
    ) {
        self.body_read_timeout_ms = core_runtime_conf.body_read_timeout_ms;
        self.body_write_timedout_ms = core_runtime_conf.body_write_timeout_ms;
    }

    fn process_request(&mut self) -> Result<()> {
        let h_v = self.req.headers.get_header_all(HeaderName::TRAILER);
        for h in h_v {
            let iter = h.get_value().split(|c| *c == b',');
            for mut v in iter {
                v = v.trim_ascii();
                if !v.is_empty() {
                    self.expect_trailers += 1;
                }
            }
        }

        Ok(())
    }

    async fn write_resp_body_helper(
        &mut self,
        data: Bytes,
        end_stream: bool,
    ) -> Result<()> {
        let body_write_timeout_ms = self.body_write_timedout_ms;
        let write_event = self.write_all(data, end_stream);
        let write_event = timeout(
            Duration::from_millis(body_write_timeout_ms as u64),
            write_event,
        );
        write_event
            .await
            .if_err(ErrorType::WriteTimedout, "while h2 send response body")?
            .if_err(ErrorType::WriteError, "while h2 send response body")
    }

    async fn write_all(
        &mut self,
        mut data: Bytes,
        end_stream: bool,
    ) -> Result<()> {
        loop {
            if data.is_empty() {
                return Ok(());
            }

            let body_writer = self.body_writer.as_mut().unwrap();
            body_writer.reserve_capacity(data.len());

            let n = match poll_fn(move |cx| body_writer.poll_capacity(cx)).await
            {
                Some(Ok(0)) => {
                    return Error::e_explain(
                        ErrorType::H2StreamClose,
                        "poll_capacity return zero bytes",
                    );
                }
                Some(Ok(n)) => n,

                Some(Err(e)) => {
                    return Err(Error::from_error(
                        ErrorType::H2StreamError,
                        "poll_capacity return error",
                        e,
                    ));
                }

                None => {
                    return Error::e_explain(
                        ErrorType::H2StreamClose,
                        "poll_capacity return none",
                    );
                }
            };

            let todo = n.min(data.len());
            let body_writer = self.body_writer.as_mut().unwrap();

            if todo == data.len() {
                return body_writer
                    .send_data(data, false)
                    .if_err(ErrorType::H2Error, "send_data error");
            }

            let new_data = data.split_to(todo);
            body_writer
                .send_data(new_data, end_stream)
                .if_err(ErrorType::H2StreamError, "send_split_data error")?;
        }
    }
}
